package com.hoshiicloud.saleorder.mq;

import com.hoshiicloud.common.exception.CommonError;
import com.hoshiicloud.common.exception.ServiceException;
import java.util.List;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * 消息消费者
 */
//@SpringBootConfiguration
public class RocketMqConsumer {

    //    @Value("${spring.rocketmq.namesrvAddr}")
    private String namesrvAddr;

    //    @Value("${spring.rocketmq.producerGroupName}")
    private String groupName;

    //    @Value("${spring.rocketmq.consumeThreadMin}")
    private int consumeThreadMin;

    //    @Value("${spring.rocketmq.consumeThreadMax}")
    private int consumeThreadMax;

    //    @Value("${spring.rocketmq.topics}")
    private String topics;
    //    @Value("${spring.rocketmq.consumeMessageBatchMaxSize}")
    private int consumeMessageBatchMaxSize;

    //    @Bean
    public DefaultMQPushConsumer defaultMQPushConsumer() {
        if (StringUtils.isEmpty(groupName)) {
            throw new ServiceException(CommonError.error("groupName is null !!!"));
        }
        if (StringUtils.isEmpty(namesrvAddr)) {
            throw new ServiceException(CommonError.error("namesrvAddr is null !!!"));
        }
        if (StringUtils.isEmpty(topics)) {
            throw new ServiceException(CommonError.error("topics is null !!!"));
        }
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                    ConsumeOrderlyContext consumeOrderlyContext) {
                if (CollectionUtils.isEmpty(msgs)) {
                    System.out.println("接受到的消息为空，不处理，直接返回成功");
                    return ConsumeOrderlyStatus.SUCCESS;
                }
                for (MessageExt m : msgs) {
                    System.out.println(
                            "接受到的消息topic为：" + m.getTopic() + "，tags为" + m.getTags() + ",body为"
                                    + new String(m.getBody()));
                }
//        if(messageExt.getTopic().equals("你的Topic")){
//            if(messageExt.getTags().equals("你的Tag")){
//                //TODO 判断该消息是否重复消费（RocketMQ不保证消息不重复，如果你的业务需要保证严格的不重复消息，需要你自己在业务端去重）
//                //TODO 获取该消息重试次数
//                int reconsume = messageExt.getReconsumeTimes();
//                if(reconsume ==3){//消息已经重试了3次，如果不需要再次消费，则返回成功
//                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//                }
//                //TODO 处理对应的业务逻辑
//            }
//        }
                // 如果没有return success ，consumer会重新消费该消息，直到return success
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
         * 如果非第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        /**
         * 设置消费模型，集群还是广播，默认为集群
         */
        //consumer.setMessageModel(MessageModel.CLUSTERING);
        /**
         * 设置一次消费消息的条数，默认为1条
         */
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        try {
            /**
             * 设置该消费者订阅的主题和tag，如果是订阅该主题下的所有tag，则tag使用*；如果需要指定订阅该主题下的某些tag，则使用||分割，例如tag1||tag2||tag3
             */
            String[] topicTagsArr = topics.split(";");
            for (String topicTags : topicTagsArr) {
                String[] topicTag = topicTags.split("~");
                consumer.subscribe(topicTag[0], topicTag[1]);
            }
            consumer.start();
            System.out.println("consumer is start !!! groupName:{},topics:{},namesrvAddr:{}");
        } catch (MQClientException e) {
            System.out.println("consumer is start !!! groupName:{},topics:{},namesrvAddr:{}");
            throw new ServiceException(CommonError.error("出现异常"));
        }
        return consumer;
    }
}
